package mq

import (
	"context"
	"fmt"
	"log"
	"strings"
	"sync"
	"time"

	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
)

// RocketMQPlugin 实现基于 RocketMQ 的消息队列
type RocketMQPlugin struct {
	nameServers     []string
	groupID         string
	closed          bool
	startTime       int64
	errorCount      int
	lastError       string
	queueStats      map[string]*QueueStats
	statsMutex      sync.RWMutex
	consumersMutex  sync.Mutex
	activeConsumers map[string]rocketmq.PushConsumer
}

// 确保 RocketMQPlugin 实现了 MQPlugin 接口
var _ MQPlugin = (*RocketMQPlugin)(nil)

// newRocketMQPlugin 创建一个新的 RocketMQPlugin 实例
func newRocketMQPlugin(config map[string]interface{}) (MQPlugin, error) {
	// 从配置中获取 NameServer 地址
	nameServersStr, ok := config["name_servers"].(string)
	if !ok || nameServersStr == "" {
		return nil, fmt.Errorf("RocketMQ NameServer 地址配置缺失或无效")
	}

	// 分割多个 NameServer 地址
	nameServers := strings.Split(nameServersStr, ",")
	for i, ns := range nameServers {
		nameServers[i] = strings.TrimSpace(ns)
	}

	// 从配置中获取 GroupID
	groupID, ok := config["group_id"].(string)
	if !ok || groupID == "" {
		groupID = "default_group" // 使用默认值
	}

	return &RocketMQPlugin{
		nameServers: nameServers,
		groupID:     groupID,
		closed:      false,
		startTime:   time.Now().Unix(),
		queueStats:  make(map[string]*QueueStats),
	}, nil
}

// Type 返回消息队列类型
func (p *RocketMQPlugin) Type() MQType {
	return RocketMQ
}

// Publish 发布消息到指定主题
func (p *RocketMQPlugin) Publish(ctx context.Context, topic string, message string) error {
	// 创建生产者
	prod, err := rocketmq.NewProducer(
		producer.WithNameServer(p.nameServers),
		producer.WithGroupName(p.groupID),
		producer.WithRetry(2),
	)
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("创建 RocketMQ 生产者失败: %v", err)
		return fmt.Errorf("创建 RocketMQ 生产者失败: %w", err)
	}

	// 启动生产者
	if err := prod.Start(); err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("启动 RocketMQ 生产者失败: %v", err)
		return fmt.Errorf("启动 RocketMQ 生产者失败: %w", err)
	}
	defer prod.Shutdown()

	// 创建消息
	msg := primitive.NewMessage(topic, []byte(message))

	// 发送消息
	res, err := prod.SendSync(ctx, msg)
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("向 RocketMQ 发送消息失败: %v", err)
		return fmt.Errorf("向 RocketMQ 发送消息失败: %w", err)
	}

	log.Printf("发送消息成功，消息ID: %s", res.MsgID)
	return nil
}

// Subscribe 订阅指定主题的消息
func (p *RocketMQPlugin) Subscribe(ctx context.Context, topic string, handler func(message string) error) error {
	// 初始化或更新队列统计信息
	p.statsMutex.Lock()
	if _, exists := p.queueStats[topic]; !exists {
		p.queueStats[topic] = &QueueStats{
			TopicName: topic,
		}
	}
	p.statsMutex.Unlock()

	// 创建消费者
	c, err := rocketmq.NewPushConsumer(
		consumer.WithNameServer(p.nameServers),
		consumer.WithGroupName(p.groupID),
	)
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("创建 RocketMQ 消费者失败: %v", err)
		return fmt.Errorf("创建 RocketMQ 消费者失败: %w", err)
	}

	// 订阅主题
	err = c.Subscribe(topic, consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		for _, msg := range msgs {
			// 更新统计信息
			p.statsMutex.Lock()
			stats := p.queueStats[topic]
			stats.LastConsumedAt = time.Now().Unix()
			stats.ConsumedMessages++
			p.statsMutex.Unlock()

			if err := handler(string(msg.Body)); err != nil {
				log.Printf("处理消息失败: %v", err)

				// 更新失败统计
				p.statsMutex.Lock()
				stats := p.queueStats[topic]
				stats.FailedMessages++
				p.statsMutex.Unlock()

				return consumer.ConsumeRetryLater, err
			}
		}
		return consumer.ConsumeSuccess, nil
	})
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("订阅 RocketMQ 主题失败: %v", err)
		return fmt.Errorf("订阅 RocketMQ 主题失败: %w", err)
	}

	// 启动消费者
	if err := c.Start(); err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("启动 RocketMQ 消费者失败: %v", err)
		return fmt.Errorf("启动 RocketMQ 消费者失败: %w", err)
	}

	// 监听上下文取消
	go func() {
		<-ctx.Done()
		c.Shutdown()
	}()

	return nil
}

// Close 关闭 RocketMQ 连接
func (p *RocketMQPlugin) Close() error {
	p.closed = true
	return nil
}

// GetStatus 获取RocketMQ的状态
func (p *RocketMQPlugin) GetStatus() MQStatus {
	// 尝试创建一个临时生产者来检查连接状态
	isConnected := false

	prod, err := rocketmq.NewProducer(
		producer.WithNameServer(p.nameServers),
		producer.WithGroupName(p.groupID),
	)

	if err == nil {
		err = prod.Start()
		if err == nil {
			isConnected = true
			prod.Shutdown()
		} else {
			p.errorCount++
			p.lastError = fmt.Sprintf("启动 RocketMQ 生产者失败: %v", err)
		}
	} else {
		p.errorCount++
		p.lastError = fmt.Sprintf("创建 RocketMQ 生产者失败: %v", err)
	}

	return MQStatus{
		IsConnected:    isConnected,
		ConnectionInfo: strings.Join(p.nameServers, ","),
		StartTime:      p.startTime,
		ErrorCount:     p.errorCount,
		LastError:      p.lastError,
	}
}

// GetQueueStats 获取指定主题的队列统计信息
func (p *RocketMQPlugin) GetQueueStats(topic string) (QueueStats, error) {
	p.statsMutex.RLock()
	defer p.statsMutex.RUnlock()

	// 检查队列是否存在
	stats, exists := p.queueStats[topic]
	if !exists {
		return QueueStats{}, fmt.Errorf("主题 %s 不存在或未被监控", topic)
	}

	// 由于 RocketMQ Go 客户端的限制，我们无法直接获取待处理消息数
	// 这里只返回已有的统计信息，不尝试获取实时数据
	statsCopy := *stats

	// 注意：如果需要获取实时队列信息，可能需要使用 RocketMQ 的 HTTP 管理接口
	// 或者通过 JMX 等方式获取，这超出了当前客户端库的能力范围

	return statsCopy, nil
}

// ListQueueStats 获取所有主题的队列统计信息
func (p *RocketMQPlugin) ListQueueStats() []QueueStats {
	p.statsMutex.RLock()
	defer p.statsMutex.RUnlock()

	result := make([]QueueStats, 0, len(p.queueStats))
	for _, stats := range p.queueStats {
		result = append(result, *stats)
	}

	return result
}

// GetPendingMessages 获取待处理的消息
func (p *RocketMQPlugin) GetPendingMessages(topic string, page, pageSize int) ([]string, error) {
	// 实现获取待处理消息的逻辑
	return []string{}, nil
}

// GetConsumedMessages 获取已消费的消息
func (p *RocketMQPlugin) GetConsumedMessages(topic string, page, pageSize int) ([]MessageRecord, error) {
	// 实现获取已消费消息的逻辑
	return []MessageRecord{}, nil
}

// GetFailedMessages 获取失败的消息
func (p *RocketMQPlugin) GetFailedMessages(topic string, page, pageSize int) ([]MessageRecord, error) {
	// 实现获取失败消息的逻辑
	return []MessageRecord{}, nil
}

// CloseQueue 关闭指定主题的队列
func (p *RocketMQPlugin) CloseQueue(topic string) error {
	// 检查连接状态
	if p.closed {
		return fmt.Errorf("RocketMQ 连接已关闭")
	}

	// 从活跃消费者中查找并关闭
	p.consumersMutex.Lock()
	defer p.consumersMutex.Unlock()

	consumer, exists := p.activeConsumers[topic]
	if !exists {
		// 如果没有活跃的消费者，只从统计信息中移除
		p.statsMutex.Lock()
		delete(p.queueStats, topic)
		p.statsMutex.Unlock()
		return nil
	}

	// 关闭消费者
	err := consumer.Shutdown()
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("关闭 RocketMQ 消费者失败: %v", err)
		return fmt.Errorf("关闭 RocketMQ 消费者失败: %w", err)
	}

	// 从活跃消费者映射中移除
	delete(p.activeConsumers, topic)

	// 从统计信息中移除
	p.statsMutex.Lock()
	delete(p.queueStats, topic)
	p.statsMutex.Unlock()

	return nil
}

// ListTopics 获取所有主题列表
func (p *RocketMQPlugin) ListTopics() []string {
	// RocketMQ没有直接的API来列出所有主题
	// 返回已知的主题（已经被监控的队列）
	p.statsMutex.RLock()
	defer p.statsMutex.RUnlock()

	topics := make([]string, 0, len(p.queueStats))
	for topic := range p.queueStats {
		topics = append(topics, topic)
	}

	return topics
}
